package com.leigq.www.producer.task;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.leigq.www.common.entity.Order;
import com.leigq.www.common.enums.KafKaMsgBizType;
import com.leigq.www.producer.constant.MessageLogConstant;
import com.leigq.www.producer.entity.BrokerMessageLog;
import com.leigq.www.producer.enumeration.MessageLogStatusEnum;
import com.leigq.www.producer.sender.KafKaSender;
import com.leigq.www.producer.sender.KafKaSenderFactory;
import com.leigq.www.producer.service.BrokerMessageLogService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * KafKa相关任务
 *
 * @author leiguoqing
 */
@Slf4j
@Component
public class KafKaTask {

    @Autowired
    private ObjectMapper objectMapper;

    @Autowired
    private BrokerMessageLogService messageLogService;

    @Autowired
    private KafKaSenderFactory kafKaSenderFactory;

    /**
     * KafKa消息重投任务，用于处理投递到KafKa失败的消息的重新投递
     * <br/>
     * 重投消息规则：每隔一定时间去数据库查询 状态为确认中 并且 下次重试时间 <= 当前时间 进行重投
     */
    @SuppressWarnings(value = "unchecked")
    @Scheduled(cron = "*/10 * * * * ?")
    public void rethrow() {
        log.info("重投KafKa消息任务开始执行......");

        // 查询 状态为确认中 并且 下次重试时间 <= 当前时间 进行重投
        List<BrokerMessageLog> logs = messageLogService.findOvertimeMsg();

        // 循环重投
        logs.forEach(messageLog -> {
            // 尝试次数 >=3 次将状态改为失败，此时需要手动检查 MQ 的状态
            if (messageLog.getTryCount() >= MessageLogConstant.TRY_COUNT) {
                messageLog.setStatus(MessageLogStatusEnum.FAIL.getValue());
                messageLogService.update(messageLog);
            } else {
                // 开始重投
                try {
                    /* 发送消息至KafKa */
                    final KafKaMsgBizType bizType = KafKaMsgBizType.getByCode(messageLog.getBizType());
                    final KafKaSender<Order> kafKaSender = kafKaSenderFactory.getSender(bizType);
                    kafKaSender.send(objectMapper.readValue(messageLog.getMessage(), Order.class));
                } catch (Exception e) {
                    log.error("发送订单消息至KafKa异常", e);
                }
                // 重投次数 + 1
                messageLog.setTryCount(messageLog.getTryCount() + 1);
                messageLogService.update(messageLog);
            }
        });

    }

}
